What is the purpose of this change?

Flink RowData carries a RowKind header (+I / -U / +U / -D, byte ordinals 0/1/2/3) that distinguishes changelog records. Today the RowKind signal is dropped at the Flink→velox boundary, so every row exiting the stateful pipeline is +I regardless of the original changelog marker. This is the velox-side foundation for threading RowKind through the stateful operator chain.

What changes are proposed in this pull request?

RowKind lives on StreamRecord directly — no new wrapper class. The merged RowVector convention (trailing $row_kind TINYINT column, constant kRowKindColumnName in velox/experimental/stateful/RowKind.h) is the wire format across the JNI boundary; it is split/recombined at the StreamRecord layer.

• StreamRecord (velox/experimental/stateful/StreamElement.h):
  • New field rowKind_ : SimpleVectorPtr<int8_t>. nullptr means append-only.
  • appendOnly() returns !rowKind_.
  • toMergedRowVector(alwaysAppendRowKind = false) returns N-column RowVector (appendOnly) or (N+1)-column RowVector with trailing $row_kind TINYINT (changelog). When alwaysAppendRowKind is true, appendOnly records get a ConstantVector<int8_t>(INSERT) trailing column — used by callers that need a fixed N+1 schema.
  • Static factory StreamRecord::create(nodeId, merged) splits a merged RowVector back into record + rowKind. A trailing ConstantVector<int8_t>(INSERT) is normalized back to appendOnly (rowKind = nullptr) so sources that always emit the column don't force every downstream operator onto the changelog path.
• StatefulSourceOperator (new header): wraps a source operator (TableScan, etc.). StreamRecord::create is used to interpret whatever the source emits — a RowVector with a trailing $row_kind column yields a per-row RowKind StreamRecord; a plain RowVector yields an appendOnly StreamRecord. No assumption that "sources only emit INSERT" — CDC-style sources (Pulsar/Kafka CDC, ORC tables with metadata columns) can emit a $row_kind column directly.
• StreamPartition: carry input rowKind alongside the RowVector through the partitioner.

Related design + phased plan: support-rowkind-column-issue

Test

New C++ unit test velox_stateful_changelog_row_vector_test